package org.cary.lucene;

import java.util.ArrayList;
import java.util.concurrent.BlockingQueue;


public abstract class Consumer<E> implements Runnable {
	
	private long batchNum;
	protected final BlockingQueue<E> queue;
	protected ArrayList<E> list = new ArrayList<E>();
	//private Logger log = Logger.getLogger(Consumer.class);
	
	public Consumer(BlockingQueue<E> queue, int batchNum ) {
		this.queue = queue;
		this.batchNum = batchNum;
	}
	
	@Override
	public void run() {		
		try {
			// 自动消费
			while (true) {
				consume(queue.take());
			}
		} catch (InterruptedException ex) {
			//log.error(ex.getMessage(), ex);
			ex.printStackTrace();
		} catch (Exception ex){
			//log.error(ex.getMessage(), ex);
			ex.printStackTrace();
		}
	}

	/**
	 * 消费数据，把数据放在list中，便于对数据的批量写入
	 * @param message
	 */
	private void consume(E record) {
		synchronized(list) {
			list.add(record);
			if(list.size() == batchNum){
				insert();
			}
		}		
	}
	protected abstract void insert();
}

